package consumer

import (
	"context"
	"fmt"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"listenboos-common/kafka/model"
	"search/dao"
	"search/kafka/saramax"
)

type AlbumConsumer struct {
	client         sarama.Client
	albumSearchDao *dao.AlbumElasticDAO
}

func NewChAlbumConsumer(
	client sarama.Client,
) *AlbumConsumer {
	return &AlbumConsumer{
		client:         client,
		albumSearchDao: dao.NewAlbumElasticDAO(),
	}
}
func (r *AlbumConsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("albumInfo",
		r.client)
	if err != nil {
		zap.L().Error("im系统kafka消息记录消费错误", zap.Error(err))
		return err
	}
	go func() {
		err = cg.Consume(context.Background(),
			[]string{"albumInfo"},
			saramax.NewHandler[model.AlbumInfo](r.AlbumConsume))
		if err != nil {
			// 记录日志
			zap.L().Error("im系统kafka消息记录消费错误", zap.Error(err))
			//	return
		}
	}()
	return err
}
func (r *AlbumConsumer) AlbumConsume(msg *sarama.ConsumerMessage, t model.AlbumInfo) error {
	// 更新es
	fmt.Println("插入es中")
	ctx := context.Background()
	err := r.albumSearchDao.InputAlbum(ctx, t)
	if err != nil {
		zap.L().Error("Consume Insert error:", zap.Error(err))
		return err
	}
	// 需要更新一下消息总数
	return nil
}
func NewChatConsumers(consumer *AlbumConsumer) []Consumer {
	KafkaConsumers = append(KafkaConsumers, consumer)
	return KafkaConsumers
}
